package com.tensai.study.framework.reactornetty.demo;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;

import java.time.Duration;

@Slf4j
public class MyServerConnection {

  final Connection conn;

  public MyServerConnection(Connection conn) {
    this.conn = conn;
  }

  public void handle() {
    conn.inbound().receiveObject() // 从StringToIntegerDecoder接收到作为结果接收到的对象
        .log("MyServerConnection")
        .delayElements(Duration.ofSeconds(1)) // 将下一个元素的处理延迟1秒
        .doOnNext(s ->
            log.info("Current received and decoded element: " + s))
        .take(5) // 在收到五个元素后取消（实际上断开了客户端的连接）
        .flatMap(s ->
            conn.outbound().sendString(
                Mono.just(String.format("byte count: %d", (Integer) s))
            ).then()
        )
        .subscribe(conn.disposeSubscriber()); // 我们必须按顺序使用该核心订阅者，以使连接在take（5）之后断开连接
  }
}
